package org.hepeng.fabric.gateway.event;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.SmartLifecycle;

import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;

/**
 * @author he peng
 * @date 2022/3/30
 */
public class ChainCodeEventListenerContainer implements SmartLifecycle {

    private static final Logger LOG = LoggerFactory.getLogger(ChainCodeEventListenerContainer.class);

    private static final int THREAD_COUNT_THRESHOLD = 8;

    private static final Map<String , ExecutorService> CHAIN_CODE_LISTEN_EXECUTOR_MAP = new ConcurrentHashMap<>();

    private volatile boolean running = false;

    private final Object lifecycleMonitor = new Object();

    private ExecutorService executorService;

    @Override
    public void start() {
        if (isRunning()) {
            return;
        }

        synchronized (this.lifecycleMonitor) {
            this.running = true;
            this.lifecycleMonitor.notifyAll();
        }

        if (! CCEventListenerRegistry.hasListener()) {
            return;
        }

        Set<String> chainCodeNameSet = CCEventListenerRegistry.getListeningChaincodes();
        int count = chainCodeNameSet.size();

        if (count > THREAD_COUNT_THRESHOLD && Objects.isNull(executorService)) {
            throw new IllegalStateException("The number of chaincodes exceeds the threshold, please manage the thread pool yourself");
        } else if (count > THREAD_COUNT_THRESHOLD) {
            for (String chainCodeName : chainCodeNameSet) {
                this.executorService.execute(new EventPoller(chainCodeName));
            }
        }

        if (count <= THREAD_COUNT_THRESHOLD) {
            for (String chainCodeName : chainCodeNameSet) {
                String threadName = chainCodeName + "-event-listening";
                ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory(threadName, false));
                CHAIN_CODE_LISTEN_EXECUTOR_MAP.put(chainCodeName , executor);

                executor.execute(new EventPoller(chainCodeName));
            }
        }


    }

    @Override
    public void stop() {
        try {
            shutdown();
        }
        finally {
            synchronized (this.lifecycleMonitor) {
                this.running = false;
                this.lifecycleMonitor.notifyAll();
            }
        }
    }

    private void shutdown() {
        synchronized (this.lifecycleMonitor) {
            if (! isRunning()) {
                LOG.info("Shutdown ignored - container is not running already");
                return;
            }
            this.running = false;
            this.lifecycleMonitor.notifyAll();
        }

        LOG.debug("Shutting down chaincode event listener container");

        for (Map.Entry<String, ExecutorService> entry : CHAIN_CODE_LISTEN_EXECUTOR_MAP.entrySet()) {
            ExecutorService executorService = entry.getValue();
            executorService.shutdown();
        }

        if (Objects.nonNull(executorService)) {
            executorService.shutdown();
        }

    }

    @Override
    public boolean isRunning() {
        synchronized (this.lifecycleMonitor) {
            return (this.running);
        }
    }

    public void setExecutor(ExecutorService executorService) {
        this.executorService = executorService;
    }
}
